package org.myfram.flink.flinkonjar.common.connector.hbase.sink;

import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.connector.hbase.sink.HBaseSinkFunction;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.Put;
import org.myfram.flink.flinkonjar.common.config.HBaseConfig;
import org.myfram.flink.flinkonjar.common.connector.hbase.HBaseTableDefinition;
import org.myfram.flink.flinkonjar.common.connector.hbase.HBaseTableMutationConverter;
import org.myfram.flink.flinkonjar.common.properties.HBaseProperties;

public class HBaseSinkBuilder<T> {
    
    private final HBaseTableDefinition<T> tableDefinition;
    private Configuration hbaseConfig;
    private long bufferFlushMaxSizeInBytes = 1024 * 1024 * 10;
    private long bufferFlushMaxMutations = 1000L;
    private long bufferFlushIntervalMillis = 1000L;
    
    public HBaseSinkBuilder(HBaseTableDefinition<T> tableDefinition, HBaseProperties prop) {
        this.tableDefinition = tableDefinition;
        this.hbaseConfig = HBaseConfig.getSourceConf(prop);
    }
    
    public HBaseSinkBuilder<T> withConfig(Configuration config) {
        this.hbaseConfig = config;
        return this;
    }
    
    public HBaseSinkFunction<T> build() {
        HBaseTableMutationConverter<T> converter = new HBaseTableMutationConverter<>(tableDefinition);
        return new HBaseSinkFunction<T>(
            tableDefinition.getTableName(),
            hbaseConfig,
            converter,
            bufferFlushMaxSizeInBytes,
            bufferFlushMaxMutations,
            bufferFlushIntervalMillis
        );
    }
}